package com.yomob.kylin.service;

import com.clearspring.analytics.util.Lists;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.UpdateManyModel;
import com.mongodb.client.model.UpdateOptions;
import com.yomob.kylin.client.MongoDBClient;
import com.yomob.kylin.facility.Consts;
import com.yomob.kylin.support.OptionSupport;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.storage.StorageLevel;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.stream.Collectors;

import static com.mongodb.client.model.Filters.eq;
import static org.apache.spark.sql.functions.col;

/**
 * Just collect data from ADX clicked counter
 * Collect ad scene,ad type,device orientation
 */
public class AdClickedService extends SimpleService{
    private static final Object[] COUNTERIDS = {"ad_adclick", "cp_adclick"};
    private static final Logger LOGGER = LoggerFactory.getLogger(AdClickedService.class);
    private static final String COLLECTION_NAME = "daily_user_info";
    private MongoDBClient mongoDBClient;
    public AdClickedService(MongoDBClient mongoDBClient) {
        this.mongoDBClient = mongoDBClient;
    }
    /**
     * Running Entry for ad clicked collect service
     * @param input,spark data set during the start-day and end-day
     * @throws Exception
     */
    @Override
    public void execute(Dataset<Row> input) throws Exception {
        Dataset<Row> dataset = input.filter(col(Consts.Fields.COUNTERID).isin("ad_adclick", "cp_adclick")).select(
                Consts.Fields.UDID,
                Consts.Fields.EXT$ADTYPE,
                Consts.Fields.EXT$ADID,
                // todo
                //"orientation",//adx is orientation，屏幕方向
                Consts.Fields.EXT$ADSCENE,
                Consts.Fields.DATE
        );

        dataset.persist(StorageLevel.fromString(OptionSupport.getStorageLevel().name()));
        List<Row> rowList = dataset.collectAsList();
        List<UpdateManyModel<Document>> models = Lists.newArrayList();
        models.addAll(rowList.stream().map(row -> this.convertClick(row)).collect(Collectors.toList()));
        mongoDBClient.saveToMongoBulk(models, COLLECTION_NAME);
        dataset.unpersist();
    }
    /**
     * return this service deal counters type
     * @return String array
     */
    @Override
    public Object[] getCounterids() {
        return COUNTERIDS ;
    }
    /**
     * Private method,used to convert spark row data to mongo update command set,contain $set, $push,and $addToSet.
     * Row data may not come from adx,used weather contains "video_url_md5" for recognised.
     * @param row
     * @return mongodb UpdateManyModel<Document> for batch update
     */
    private UpdateManyModel<Document> convertClick(Row row) {
        List<Bson> matches = Lists.newArrayList();
        //base on device id to update
        Bson eq = eq(Consts.UserInfoBased.DEVICE_ID, row.getAs(Consts.Fields.UDID));
        matches.add(eq);
        Bson filter = Filters.and(matches);
        Document update = new Document();
        Document set = new Document();
        String adType = row.getAs("adtype");
        if(!row.isNullAt(2)){
            String adid = row.getAs("video_url_md5");
            set.put(Consts.UserInfoBehavior.AD_ID,adid);
            Document document = new Document(Consts.UserInfoBehavior.CLICKED_AD_IDS,adid);
            update.append(Consts.MongoCommand.PUSH,document);
        }
        if(adType !=null && !adType.equals("null")){
            set.put(Consts.UserInfoBehavior.AD_TYPE,adType);
            update.append(Consts.MongoCommand.ADD_TO_SET, new Document(Consts.UserInfoBehavior.AD_TYPES, row.getAs(Consts.Fields.DATE)+"_"+adType));
        }
        set.put(Consts.UserInfoBehavior.AD_SCENE,row.getAs("adscene"));
        //update.append(Consts.UserInfoBehavior.SCREEN_DIRECTION,row.getAs("orientation"));//adx & sdk is different
        update.append(Consts.MongoCommand.SET,set);
        UpdateOptions updateOptions = new UpdateOptions().upsert(true);
        return new UpdateManyModel<>(filter, update, updateOptions);
    }
}
